package themis.flink.riskengine;

import com.alibaba.fastjson.JSONObject;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.util.Collector;

public class EventSplitFunction implements FlatMapFunction<JSONObject, JSONObject> {
    private Set<String> measureAndDimension; // amount#rcv_account.history

    public EventSplitFunction(List<Rule> rules) {
        rules.forEach(x -> this.measureAndDimension.add(x.getMeasure() + "#" + x.getDimension()));
    }

    public void flatMap(JSONObject value, Collector<JSONObject> out) throws Exception {
        String eventId = UUID.randomUUID().toString();
        long timestamp = value.getLongValue("timestamp");
        JSONObject event = new JSONObject();
        event.put("KEY_NAME", "event");
        event.put("KEY_VALUE", eventId);
        event.put("EVENT_ID", eventId);
        event.putAll((Map)value);
        out.collect(event);
        this.measureAndDimension.forEach(key -> {
            JSONObject json = new JSONObject();
            json.put("timestamp", Long.valueOf(timestamp));
            json.put("KEY_NAME", key);
            json.put("KEY_VALUE", genKeyValue(value, key));
            json.put("EVENT_ID", eventId);
//            json.put("")
//            genKeyFields(key).forEach(());// TODO
            out.collect(json);
        });
    }

    private String genKeyValue(JSONObject event, String key) {
        String keyValue;
        if (!key.endsWith(".history"))
            throw new UnsupportedOperationException("unsupported key type");
        String[] splits = key.replace(".history", "").split("#"); // amount#rcv_account
        if (splits.length == 1) {
            String target = splits[0];
            keyValue = String.format("%s#%s.history", target, event.get(target));
        } else if (splits.length == 2) {
            String target = splits[0];
            String on = splits[1];
            keyValue = String.format("%s#%s.history", target, event.get(on));
        } else {
            throw new UnsupportedOperationException("unsupported key type");
        }
        return keyValue;
    }

    private Set<String> genKeyFields(String key) {
        if (!key.endsWith(".history"))
            throw new UnsupportedOperationException("unsupported key type");
        String[] splits = key.replace(".history", "").split("#");
        return new HashSet<>(Arrays.asList(splits));
    }
}
